package com.htdata.iiot;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import com.alibaba.fastjson.JSONObject;
import com.htdata.iiot.api.pojo.DeviceNumber;
import com.htdata.iiot.api.pojo.StreamingData;
import com.htdata.iiot.api.pojo.StreamingMetricData;
import com.htdata.iiot.db.DbConnectionPool;
import com.htdata.iiot.db.JDBCUtils;

public class Main {
	private static DbConnectionPool pool = DbConnectionPool.getPool();
	private static String topic;
	public static void main(String[] args) {
		int tid=-1;
		int did=-1;
		int sid=-1;
		int tim=5;
		int limit=-1;
		String model="*";
		for(String arg:args){
			String[] split = arg.split("=");
			switch (split[0]) {
			case "topic":
				topic = split[1];
				Configs.KafkaConfig.TOPIC=topic;
				break;
			case "tid":
				tid = Integer.parseInt(split[1]);
				break;
			case "tim":
				tim = Integer.parseInt(split[1]);
				break;
			case "model":
				model = split[1];
				break;

			default:
				break;
			}
		}
		
		
		List<StreamingData> sList = new ArrayList<StreamingData>();
		SendDataToKafka sendDataToKafka = new SendDataToKafka();
		// 1.根据租户ID获取设备
		/*String sql1 = "SELECT tenant_id , id,serial_number FROM `thing_device` "
				+ "WHERE deleted=0 AND type=0 ";
		if(tid!=-1){
			sql1+=" and `tenant_id` = ?";
		}*/
		String sql1="";
		sql1 = "SELECT tenant_id , id,serial_number FROM `thing_device` "
				+ "WHERE deleted=0 AND type=0 ";
		if("tenant".equals(model)||tid>0){
			if(tid<=0){
				System.out.println("未指定租户id");
				System.exit(0);
			}
			sql1 = "SELECT tenant_id , id,serial_number FROM `thing_device` "
					+ "WHERE deleted=0 AND type=0 and `tenant_id` = "+tid;
		}
		if("struct".equals(model)||sid>0){
			if(sid<=0){
				System.out.println("未指定设备模板id");
				System.exit(0);
			}
			sql1 = "SELECT tenant_id , id,serial_number FROM `thing_device` "
					+ "WHERE deleted=0 AND type=0 and `struct_id` = "+sid;
		}
		if("device".equals(model)||did>0){
			if(did<=0){
				System.out.println("未指定设备id");
				System.exit(0);
			}
			sql1 = "SELECT tenant_id , id,serial_number FROM `thing_device` "
					+ "WHERE deleted=0 AND type=0 and `id` = "+did;
		}
		if("app".equals(model)){
			sql1 = "SELECT  T2.*FROM"
					+"  (SELECT"
					+"    T.*,"
					+"    CASE"
					+"      WHEN @mid = struct_id"
					+"      THEN @row := @row + 1"
					+"      ELSE @row := 1"
					+"    END rownum,"
					+"    @mid := struct_id MID"
					+"  FROM"
					+"    (SELECT"
					+"      struct_id,"
					+"      id,tenant_id,"
					+"      SUM(1)"
					+"    FROM"
					+"      thing_device"
					+"      WHERE (id IN(SELECT device_id FROM `user_app` WHERE deleted=0 "
					+ (tid<=0?"":"AND tenant_id="+tid)
					+ " )"
					+" OR struct_id IN(SELECT device_id FROM `user_app` WHERE deleted=0 "
					+ (tid<=0?"":"AND tenant_id="+tid)
					+ " )) AND deleted=0 AND TYPE=0"
					+ (tid<=0?"":"AND tenant_id= "+tid)
					+"    GROUP BY struct_id,id,tenant_id"
					+"    ORDER BY struct_id,id,"
					+"      SUM(1) DESC) AS T) AS T2"
					+" WHERE T2.rownum <= 8100 "
					+ "ORDER BY rownum "
					+ (limit<=0?"":"LIMIT 0, "+limit)
					+ "LIMIT 0,6200";
		}
			
		
		System.out.println(sql1);
		Connection conn = pool.getConnection();
		PreparedStatement ps = null;
		ResultSet rs = null;
		ResultSet executeQuery = null;
		PreparedStatement prepareStatement = null;
		try {
			ps = conn.prepareStatement(sql1);
			/*if(tid!=-1){
				ps.setInt(1, tid);
			}*/
			rs = ps.executeQuery();
			while (rs.next()) {
				int deviceId = rs.getInt("id");
				int tenant_id = rs.getInt("tenant_id");

				StreamingData streamingData = new StreamingData();
				streamingData.setDid(deviceId);
				streamingData.setTid(tenant_id);
				// 2.根据设备获取测点
				String sql2 = "SELECT "
						+ "t1.id AS id,"
						+ "t2.`name` AS NAME "
						+ "FROM `thing_device_component_rel` t1,"
						+ " `thing_data_stream` t2 ,"
						+ "  thing_component t3 "
						+ "WHERE t1.`component_id` = t3.`id` "
						+ "  AND t3.deleted = 0 "
						+ "  AND t2.`component_id` = t3.`id` "
						+ "  AND t1.deleted = 0 "
						+ "  AND t1.device_id = ? "
						+ "  AND t2.deleted = 0 "
						+ "  AND t2.tenant_id = ?  ";
				prepareStatement = conn.prepareStatement(sql2);
				prepareStatement.setInt(1, deviceId);
				prepareStatement.setInt(2, tenant_id);
				executeQuery = prepareStatement.executeQuery();
				List<StreamingMetricData> mList = new ArrayList<StreamingMetricData>();
				while (executeQuery.next()) {
					String name = executeQuery.getString("name");
					long mId = executeQuery.getLong("id");
					StreamingMetricData metricData = new StreamingMetricData();
					metricData.setK(name);
					metricData.setDn(new DeviceNumber(tenant_id, deviceId, mId).toHexString());
					mList.add(metricData);
				}
				streamingData.setData(mList);
     				if(mList.size()>0){
     					sList.add(streamingData);
     				}
			}

		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			JDBCUtils.close(ps, rs, conn);
			JDBCUtils.close(prepareStatement, executeQuery, null);
		}
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println(sList.size());
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		System.out.println("****************************");
		long startTime = System.currentTimeMillis()-1;
		long st = System.currentTimeMillis();
		long i = 0;
		long mi = 0;
		while (true) {
			st = System.currentTimeMillis();
			// 3.生成数据
			for (StreamingData sData : sList) {
				List<StreamingMetricData> data = sData.getData();
				for (StreamingMetricData mData : data) {
					mData.setV(Math.ceil(new Random().nextDouble() * 10000) * 0.01);
					mData.setTs(st);
					mi++;
				}
				sData.setTs(st);
				sendDataToKafka.send(sData.getDid(),buildResult(sData));
				i++;
			}
			long endTime = System.currentTimeMillis();
			long useTime = endTime - startTime;
			System.out.println("写入 " + i + " 条数据,累计测点数："+mi+"，平均速率 " + (i * 1000 / useTime)+"，测点速率 " + (mi * 1000 / useTime));
			try {
				Thread.sleep(tim*1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}

		}

	}

	private static String buildResult(StreamingData sData) {
		return JSONObject.toJSONString(sData).toString();
	}

}